package com.atguigu.flink.sqlfunction;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Expressions;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.*;

/**
 * Created by Smexy on 2023/2/6
 */
public class Demo1_FunctionUse
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        String createTableSql = "create table t1( id string,ts bigint ,vc int ) with(" +
            "                    'connector' = 'filesystem' ," +
            "                    'path' = 'data/func/sensor.txt' , " +
            "                    'format' = 'csv'  " +
            "                    )";

        //执行建表(写操作)
        tableEnvironment.executeSql(createTableSql);

        //把t1变为Table对象
        Table table = tableEnvironment.from("t1");

        /*
                flink提供的系统内置函数:
                        https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/dev/table/functions/systemfunctions/
                            RAND()
                            RAND(INT)
         */

        //TableAPI使用函数
      /*  table
              //.select($("id"), rand())
              //如何向函数中传参
              .select($("id"), call("rand",$("vc")))
             .execute()
             .print();*/

        //SQL中用函数
        tableEnvironment.sqlQuery("select id,rand(vc) from t1")
                        .execute()
                        .print();


    }
}
